package com.xiaotao.learn.thread;

import java.util.LinkedList;

public class Main {
    public static void main(String[] args) throws InterruptedException {
        MessageQueue<String> mq = new MessageQueue<>();


        // 开1个线程消费消息
        new Thread(() -> {
           while (true) {
               String msg = mq.consume();
               System.out.println("消费消息：" + msg);
               if (msg == null) {
                   System.out.println("队列关闭");
                   break;
               }
           }
        }).start();
        for (int i = 0; i < 10; i++) {
            mq.product("" + i);
            System.out.println("生产消息：" + i);
            Thread.sleep(100);
        }
        mq.stop();
        Thread.sleep(1000);
        System.out.println("主线程结束");
    }
}

/**
 * 使用锁实现的消息队列
 */
class MessageQueue<T> {
    // 队列关闭标志
    private volatile boolean stop = false;
    // 使用链表作为消息容器
    private final LinkedList<T> queue = new LinkedList<>();
    /**
     * 生产消息
     * @param msg   消息内容
     */
    public void product(T msg) {
        // 获取链表的锁之后，向链表添加消息，最后尝试唤醒链表对象（如果在等待的话）
        synchronized (this) {
            if (stop) {
                throw new IllegalStateException("消息队列已关闭");
            }
            queue.add(msg);
            this.notify();
        }
    }
    /**
     * 关闭消息队列，并唤醒消费者的等待（如果在等待的话）
     */
    public void stop() {
        synchronized (this) {
            stop = true;
            this.notify();
        }
    }
    /**
     * 消费消息，若无消息将会一直阻塞，直到有消息或队列关闭，若消息队列已关闭则不再阻塞且返回null
     * @return 拿到的消息
     */
    public T consume() {
        // 获取对象的锁
        synchronized (this) {
            if (queue.size() != 0) {
                return queue.remove();
            } else {
                while (!stop && queue.size() == 0) {
                    try {
                        this.wait();
                    } catch (InterruptedException ignored) { }
                }

                // 由于关闭队列也有可能退出上面的循环，所以要再判断一次长度
                if (queue.size() == 0) {
                    return null;
                } else {
                    return queue.remove();
                }
            }
        }
    }
}
